package com.demo.kafka;

import com.zte.conn.parser.base.protobuf.OriginalData;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.utils.Utils;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.io.BufferedWriter;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;

@SpringBootApplication
public class DebugKafkaApplication implements CommandLineRunner {

    public static final String KAFKA_BOOTSTRAP_SERVER = "kafka.bootstrap.server";
    public static final String TOPIC = "topic";
    public static final String DEVICE = "device";
    public static final String TIME = "time";

    public static DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    //10.94.44.107:9092,10.94.44.108:9092,10.94.44.109:9092

    private final static char[] hexArray = "0123456789ABCDEF".toCharArray();

    public static String bytesToHex(byte[] bytes) {
        char[] hexChars = new char[bytes.length * 2];
        for ( int j = 0; j < bytes.length; j++ ) {
            int v = bytes[j] & 0xFF;
            hexChars[j * 2] = hexArray[v >>> 4];
            hexChars[j * 2 + 1] = hexArray[v & 0x0F];
        }
        return new String(hexChars);
    }


    public static void main(String[] args) {
        SpringApplication.run(DebugKafkaApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {


        Map<String,String> keyValues = new HashMap<>();

        for(String arg : args) {
            String[] values = arg.split("=");
            if(values.length == 2) {
                String key = values[0].trim();
                String value = values[1].trim();
                if(!key.isEmpty() && !value.isEmpty()) {
                    keyValues.put(key, value);
                }
            }
        }

        if(!keyValues.containsKey(KAFKA_BOOTSTRAP_SERVER)) {
            System.out.println("please specify kafka.bootstrap.server arg");
            return;
        }

        if(!keyValues.containsKey(TOPIC)) {
            System.out.println("please specify topic arg");
            return;
        }

        if(!keyValues.containsKey(DEVICE)) {
            System.out.println("please specify device arg");
            return;
        }


        if(!keyValues.containsKey(TIME)) {
            System.out.println("please specify time arg");
            return;
        }


        final Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, keyValues.get(KAFKA_BOOTSTRAP_SERVER));
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());

        final Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);

        List<PartitionInfo> partitionInfos = consumer.partitionsFor(keyValues.get(TOPIC));

        int partition = Utils.toPositive(Utils.murmur2(keyValues.get(DEVICE).getBytes())) % partitionInfos.size();

        TopicPartition topicPartition = new TopicPartition(keyValues.get(TOPIC), partition);

        consumer.assign(Collections.singletonList(topicPartition));

        Map<TopicPartition, Long> timestampsToSearch = new HashMap<>(1);

        timestampsToSearch.put(topicPartition, Long.parseLong(keyValues.get(TIME)));

        Map<TopicPartition, OffsetAndTimestamp> outOffsets = consumer.offsetsForTimes(timestampsToSearch);

        consumer.seek(topicPartition, outOffsets.get(topicPartition).offset());

        BufferedWriter bufferedWriter = Files.newBufferedWriter(Paths.get("out.txt"), StandardOpenOption.WRITE , StandardOpenOption.CREATE , StandardOpenOption.TRUNCATE_EXISTING);

        int emptyCount = 0;

        do {
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(1));

            if (!records.isEmpty()) {
                for (ConsumerRecord<String, byte[]> record : records) {

                    if (keyValues.get(DEVICE).equals(record.key())) {
                        emptyCount = 0;
                        OriginalData.originalData originalData = OriginalData.originalData.parseFrom(record.value());

                        LocalDateTime ld = Instant.ofEpochMilli(record.timestamp()).atZone(ZoneId.systemDefault()).toLocalDateTime();
                        String valueStr = bytesToHex(originalData.getOriginalData().toByteArray());

                        String line = String.format("%s %s %s", FORMATTER.format(ld), record.key(), valueStr);
                        System.out.println(line);
                        bufferedWriter.write(line);
                        bufferedWriter.newLine();
                        bufferedWriter.flush();
                    } else {
                        ++emptyCount;
                    }
                }
            } else {
                ++emptyCount;
            }

        } while (emptyCount < 10);
        System.out.println("consumer end");

        consumer.close();

        bufferedWriter.flush();
        bufferedWriter.close();

    }
}
